package com.zyx.flinkdemo.sql.time;

import com.zyx.flinkdemo.sql.cons.CommonConfig;
import com.zyx.flinkdemo.sql.utils.ConnectUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author zyx
 * @since 2021/5/29 15:36
 * desc: 通过DDL语句指定 处理时间
 */
public class ProcessTimeByDdl {
    public static void main(String[] args) {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 使用DDL语句指定处理时间
        String fields = "`id` string,\n" +
                "`ts` bigint,\n" +
                "`vc` int,\n" +
                // `pv` TIMESTAMP_LTZ(3) NOT NULL *PROCTIME* AS PROCTIME()
                // TIMESTAMP_LTZ ==> TIMESTAMP(p) WITH LOCAL TIME ZONE
                "`pv` as proctime()";
        tableEnv.executeSql(ConnectUtils.getNormalKafkaSourceConnect("sensor_source", fields,
                "sensor_source", CommonConfig.KAFKA_SERVER,
                "src_consumer", "latest-offset", "csv"));
        // 打印表信息
        Table table = tableEnv.from("sensor_source");
        table.printSchema();
    }
}
